#include "sys.h"
#include "log.h"
#include "thread_pool.h"
#include "dial.h"
#include "cJSON.h"
#include "httpspeed_tactics.h"
#include "data_report.h"
#include "trans.h"
#include "devinfo.h"
#include "comm_param.h"

static int Count = 0;

#define RESET_MIN_MAX(min, max) ((min = 0),(max = 0))

static float Max_rate = 0, Min_rate = 0;
static int* Valid_bytes_ptr;
static int Valid_bytes_count;
static int Ignore_time, Total_time, Interval_time;

struct down_up_urls {
    int up_count, down_count;
    char** up;
    char** down;
};

static int _get_cur_bytes()
{
    int i = 0;
    int bytes = 0;
    for (i = 0; i < Valid_bytes_count; i++) {
        bytes += Valid_bytes_ptr[i];
    }
    return bytes;
}
static void* _cal_max_min_rate(void* arg)
{
    int i = 0;
    float cur_rate = 0;
    int startTime = 0;
    int prev_bytes, cur_bytes;
    int total_time = Total_time;

    usleep(MSEC2USEC(Ignore_time));
    prev_bytes = _get_cur_bytes();
    total_time -= Ignore_time;
    while ((total_time -= Interval_time) > 0) {
        usleep(MSEC2USEC(Interval_time));
        cur_bytes = _get_cur_bytes();
        cur_rate = (float)(cur_bytes - prev_bytes) * 8 / MSEC2SEC(Interval_time);
        Max_rate = MAX_VAL(cur_rate, Max_rate);
        Min_rate = Min_rate ? MIN_VAL(cur_rate, Min_rate) : cur_rate;
        prev_bytes = cur_bytes;
    }
}
static void _free_down_up_urls(struct down_up_urls *urls){
    int i =0;

    for(i =0; i < urls->up_count; i++){
        free(urls->up[i]);
    }
    for(i =0; i < urls->down_count; i++){
        free(urls->down[i]);
    }
    free(urls->down);
    free(urls->up);
}
static int _get_node_url(char* url, char* probeId, char* type, struct down_up_urls* urls)
{
    cJSON *root = NULL, *resp = NULL, *arr, *obj, *data;
    char* post_data;
    int ret = 1, i = 0;

    root = cJSON_CreateObject();
    cJSON_AddStringToObject(root, "speedType", type);
    cJSON_AddStringToObject(root, "probeId", probeId);

    post_data = cJSON_Print(root);
    if ((resp = http_post(url, post_data)) == NULL) {
        log_error("Get node url error");
        ret = 0;
        goto out;
    }

    obj = cJSON_GetObjectItem(resp, "status");
    if (strcmp(obj->valuestring, "1") != 0) {
        log_error("Get node url failed");
        ret = 0;
        goto out;
    }
    arr = cJSON_GetObjectItem(resp, "uploadUrlList");
    urls->up_count = cJSON_GetArraySize(arr);
    urls->up = calloc(urls->up_count, sizeof(char*));
    for (i = 0; i < urls->up_count; i++) {
        obj = cJSON_GetArrayItem(arr, i);
        data = cJSON_GetObjectItem(obj, "url");
        urls->up[i] = strdup(data->valuestring);
    }

    arr = cJSON_GetObjectItem(resp, "downloadUrlList");
    urls->down_count = cJSON_GetArraySize(arr);
    urls->down = calloc(urls->down_count, sizeof(char*));
    for (i = 0; i < urls->down_count; i++) {
        obj = cJSON_GetArrayItem(arr, i);
        data = cJSON_GetObjectItem(obj, "url");
        urls->down[i] = strdup(data->valuestring);
    }

out:
    cJSON_Delete(root);
    if (resp)
        cJSON_Delete(resp);
    free(post_data);
    return ret;
}

void* do_httpspeed_tactics(void* arg, char* nodeUrl, char* probeId)
{

    httpspeed_tactics_t* httpspeed = (httpspeed_tactics_t*)arg;
    struct httpspeedTactics* upTactics = TAILQ_FIRST(&httpspeed->up_head);
    struct httpspeedTactics* downTactics = TAILQ_FIRST(&httpspeed->down_head);
    cJSON *result = NULL, *obj = NULL, *arr = NULL;
    struct down_up_urls urls;

    dial_arg_t dial_arg = { 0 };
    dial_result_t* dial_result = NULL;
    http_result_t* httpup_result = NULL;
    http_result_t* httpdown_result = NULL;
    char tmp_str[32] = { 0 };
    char topic[32] = { 0 };
    char *post_data = NULL, reportTime[32], collectTime[32];
    uint32_t cur_netrate = 0;
    int down_success = 0, up_success = 0;
    pthread_t thid = 0;


    if(httpspeed->down_count == 0 && httpspeed->up_count == 0){
        return NULL;
    }


    memset(&urls, 0, sizeof(urls));
    if (!_get_node_url(nodeUrl, probeId, "http", &urls)) {
        return NULL;
    }

    result = cJSON_CreateObject();
    if(!result){
        log_trace("cJSON_CreateObject failed.");
        return NULL;
    }
    cJSON_AddStringToObject(result, "dialTaskId", httpspeed->taskId);
    cJSON_AddStringToObject(result, "version", httpspeed->version);
    cJSON_AddStringToObject(result, "indexType", "http");
    get_timestring(reportTime);
    cJSON_AddStringToObject(result, "reportTime", reportTime);
    get_timestring(collectTime);
    cJSON_AddStringToObject(result, "collectTime", collectTime);
    // pthread_mutex_lock(&httpspeed->mutex);
    arr = cJSON_CreateArray();
    if(!arr){
        log_trace("cJSON_CreateArray failed.");
        cJSON_Delete(result);
        return NULL;
    }
    obj = cJSON_CreateObject();
    if(!obj){
        log_trace("cJSON_CreateObject failed.");
        cJSON_Delete(result);
        cJSON_Delete(arr);
        return NULL;
    }
        
    SPEEDTEST_LOCK();

    //由于光猫建立了两条通道，对下载速度互不影响，不用检测背景流量,
    //但光猫在路由模式下则需检测背景流量。
    log_trace("check if need monitor backgroud flow.");
    if(devinfo_is_bridge() == 0 && downTactics){
        cur_netrate = devinfo_monitor_net_rate(&devInfo);
        log_info("cur_netrate:%.2f KBps, threshold:%.2f KBps", KBps(cur_netrate), KBps(downTactics->threshold * commReqParam.bdcode_val));
    }

    if(downTactics){
        if(cur_netrate > (downTactics->threshold * commReqParam.bdcode_val) && commReqParam.bdcode_val > 0 && downTactics->threshold > 0){
            sprintf(tmp_str, "%.0f", KBps(cur_netrate));
            cJSON_AddStringToObject(obj, "bgFlowRate", tmp_str);
            cJSON_AddNumberToObject(obj, "resultCode", -1);

            cJSON_AddStringToObject(obj, "downMinSpeed", "0");
            cJSON_AddStringToObject(obj, "downMaxSpeed", "0");
            cJSON_AddStringToObject(obj, "downAvgSpeed", "0");
            cJSON_AddNumberToObject(obj, "downActualDuration", 0);
            cJSON_AddNumberToObject(obj, "downActualSize", 0);
            
            cJSON_AddStringToObject(obj, "upMinSpeed", "0");
            cJSON_AddStringToObject(obj, "upMaxSpeed", "0");
            cJSON_AddStringToObject(obj, "upAvgSpeed", "0");
            cJSON_AddNumberToObject(obj, "upActualDuration", 0);
            cJSON_AddNumberToObject(obj, "upActualSize", 0);
        }
    }
    else{

        cJSON_AddNumberToObject(obj, "bgFlowRate", 0);
        cJSON_AddNumberToObject(obj, "resultCode", 0);


        if (downTactics) {
            log_trace("do httpSpeed download");
            memset(&dial_arg, 0, sizeof(dial_arg));

            RESET_MIN_MAX(Min_rate, Max_rate);
            dial_arg.in.http_arg.uris = urls.down;
            dial_arg.in.http_arg.uris_count = urls.down_count;
            dial_arg.in.http_arg.threads = downTactics->threads;
            Valid_bytes_count = dial_arg.in.http_arg.threads;
            Valid_bytes_ptr = calloc(Valid_bytes_count, sizeof(int));
            dial_arg.in.http_arg.valid_bytes_ptr = Valid_bytes_ptr;
            Total_time = dial_arg.in.http_arg.duration = SEC2MSEC(downTactics->totalTime);
            Interval_time = dial_arg.in.http_arg.interval = SEC2MSEC(1);
            Ignore_time = dial_arg.in.http_arg.ignore_duration = SEC2MSEC(downTactics->ignoreTime);
            
            pthread_create(&thid, NULL, _cal_max_min_rate, NULL);
            pthread_detach(thid);
            dial_result = (dial_result_t*)dial_http_download(&dial_arg);
            dial_result->out.http_result.min_rate = Min_rate;
            dial_result->out.http_result.max_rate = Max_rate;
            http_result_print(&dial_result->out.http_result);
    
            sprintf(tmp_str, "%.0f", KBps(dial_result->out.http_result.min_rate / 8));
            cJSON_AddStringToObject(obj, "downMinSpeed", tmp_str);
            sprintf(tmp_str, "%.0f", KBps(dial_result->out.http_result.max_rate / 8));
            cJSON_AddStringToObject(obj, "downMaxSpeed", tmp_str);
            sprintf(tmp_str, "%.0f", KBps(dial_result->out.http_result.rate / 8));
            cJSON_AddStringToObject(obj, "downAvgSpeed", tmp_str);
            cJSON_AddNumberToObject(obj, "downActualDuration", dial_result->out.http_result.total_time);
            cJSON_AddNumberToObject(obj, "downActualSize", dial_result->out.http_result.total_bytes);
            strcpy(topic, downTactics->kafkaTopic);

            if(dial_result->out.http_result.rate > 0){
                down_success = 1;
            }
            log_info("Http Download err:%d", dial_result->err);

            free(dial_result);
            free(Valid_bytes_ptr);

        }
        if (upTactics) {
            log_trace("do httpSpeed upload");

            memset(&dial_arg, 0, sizeof(dial_arg));

            RESET_MIN_MAX(Min_rate, Max_rate);
            dial_arg.in.http_arg.uris = urls.up;
            dial_arg.in.http_arg.uris_count = urls.up_count;
            dial_arg.in.http_arg.threads = upTactics->threads;
            Valid_bytes_count = dial_arg.in.http_arg.threads;
            Valid_bytes_ptr = calloc(Valid_bytes_count, sizeof(int));
            dial_arg.in.http_arg.valid_bytes_ptr = Valid_bytes_ptr;
            Total_time = dial_arg.in.http_arg.duration = SEC2MSEC(upTactics->totalTime);
            Interval_time = dial_arg.in.http_arg.interval = SEC2MSEC(1);
            Ignore_time = dial_arg.in.http_arg.ignore_duration = SEC2MSEC(upTactics->ignoreTime);
            
            pthread_create(&thid, NULL, _cal_max_min_rate, NULL);
            pthread_detach(thid);
            dial_result = (dial_result_t*)dial_http_upload(&dial_arg);
            dial_result->out.http_result.min_rate = Min_rate;
            dial_result->out.http_result.max_rate = Max_rate;
            http_result_print(&dial_result->out.http_result);

            sprintf(tmp_str, "%.0f", KBps(dial_result->out.http_result.min_rate / 8));
            cJSON_AddStringToObject(obj, "upMinSpeed", tmp_str);
            sprintf(tmp_str, "%.0f", KBps(dial_result->out.http_result.max_rate / 8));
            cJSON_AddStringToObject(obj, "upMaxSpeed", tmp_str);
            sprintf(tmp_str, "%.0f", KBps(dial_result->out.http_result.rate / 8));
            cJSON_AddStringToObject(obj, "upAvgSpeed", tmp_str);
            cJSON_AddNumberToObject(obj, "upActualDuration", dial_result->out.http_result.total_time);
            cJSON_AddNumberToObject(obj, "upActualSize", dial_result->out.http_result.total_bytes);
            
            if(dial_result->out.http_result.rate > 0){
                up_success = 1;
            }

            log_info("Http Upload err:%d", dial_result->err);
            
            free(dial_result);
            free(Valid_bytes_ptr);

        }
        _free_down_up_urls(&urls);
    }
    SPEEDTEST_UNLOCK();
    
    get_timestring(collectTime);
    cJSON_AddStringToObject(obj, "collectTime", collectTime);
    
    if(upTactics){
        cJSON_AddStringToObject(obj, "businessTag", upTactics->businessTag);
    }else if(downTactics){
        cJSON_AddStringToObject(obj, "businessTag", downTactics->businessTag);
    }
    
    
    if(up_success){
        cJSON_AddItemToArray(arr, obj);
    }else{
        cJSON_Delete(obj);
    }
    
    cJSON_AddItemToObject(result, "httpSpeedList", arr);
    post_data = cJSON_Print(result);
    if(Debug_mode()){
        log_debug("HTTPSPEED post_data:%s", post_data);
    }
    // data_report(httpspeed->kafka_url, topic, post_data, strlen(post_data) );
    free(post_data);
    // cJSON_Delete(result);
    // pthread_cond_signal(&httpspeed->cond);
    // pthread_mutex_unlock(&httpspeed->mutex);
    return (void *)result;
}